package rpcagent

import (
	"container/list"
	"errors"
	"net"
	"reflect"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	log "ac-common-go/glog"

	"git.apache.org/thrift.git/lib/go/thrift"
)

const (
	IDLECHECKINTERVAL        = 60 //清除超时连接间隔
	REMOVEDPOOLCHECKINTERVAL = 60 //移除后池对象检测间隔
)

var nowFunc = time.Now

type ThriftDial func(ip, port string, connTimeout time.Duration) (*IdleClient, error)
type ThriftClientClose func(c *IdleClient) error

type ThriftPoolManager struct {
	poolList     []*ThriftPool
	retryNum     int
	maxFailCount int64
	lock         *sync.Mutex
	poolCursor   int
}

type ThriftPool struct {
	Dial  ThriftDial
	Close ThriftClientClose

	lock        *sync.Mutex
	idle        list.List
	idleTimeout time.Duration
	connTimeout time.Duration
	maxConn     int
	initConn    int
	count       int
	ip          string
	port        string
	closed      bool
	failNum     int64
	removed     bool
}

type IdleClient struct {
	Socket     *thrift.TSocket
	Client     interface{}
	CreateTime time.Time
	destroy    bool
}

//error
var (
	ErrOverMax          = errors.New("over max connect num")
	ErrInvalidConn      = errors.New("client is nil when close")
	ErrPoolClosed       = errors.New("pool is already closed")
	ErrSocketDisconnect = errors.New("socket is already closed")
	ErrInvalidParam     = errors.New("param is invalid")
	ErrInvalidResult    = errors.New("result is invalid")
	ErrNoAvaliblePool   = errors.New("no avalible pool")
	ErrPoolManagerErr   = errors.New("pool manager is not avalible")
)

func NewThriftPoolManager(ips, port string,
	maxConn, initConn, connTimeout, idleTimeout int, dial ThriftDial,
	closeFunc ThriftClientClose, retryNum, maxFailCount int) *ThriftPoolManager {
	pm := &ThriftPoolManager{retryNum: retryNum, maxFailCount: int64(maxFailCount), lock: new(sync.Mutex)}
	ipArray := strings.Split(ips, "*")
	pm.poolList = make([]*ThriftPool, len(ipArray))
	for idx := 0; idx < len(ipArray); idx++ {
		ip := ipArray[idx]
		thriftPool := &ThriftPool{
			Dial:        dial,
			Close:       closeFunc,
			ip:          ip,
			port:        port,
			lock:        new(sync.Mutex),
			maxConn:     maxConn,
			idleTimeout: time.Duration(idleTimeout) * time.Second,
			connTimeout: time.Duration(connTimeout) * time.Second,
			closed:      false,
			count:       0,
			initConn:    initConn,
			removed:     false,
		}
		err := thriftPool.InitPool(false)
		if err != nil {
			log.Errorf("thrift pool init error")
			thriftPool.removed = true
		}
		pm.poolList[idx] = thriftPool
	}
	log.Infof("thrift pool init over:%d",  len(pm.poolList))
	go pm.StartIdleConnClearTask()
	go pm.CheckRemovedPool()
	return pm
}
func (pm *ThriftPoolManager) AskRpc(methodName string, args []interface{}) (interface{}, error) {
	if pm == nil {
		return nil, ErrPoolManagerErr
	}
	var retResult interface{}
	var retErr error

	for i := 0; i <= pm.retryNum; i++ {
		pool := pm.GetPoolElement()
		if pool == nil {
			return nil, ErrNoAvaliblePool
		}
		retResult, retErr = pool.SendRpc(methodName, args)
		if retErr == nil {
			pool.resetFaileNum()
			break
		}
		if retErr == ErrInvalidParam || retErr == ErrOverMax {
			return nil, retErr
		} else {
			pool.incrFaileNum()
			if pool.failNum >= pm.maxFailCount {
				log.Errorf("pool fail num is over max, will be removed")
				pm.RemovePoolElement(pool)
			}
			log.Errorf("idle client send rpc error:%+v, retry:%d", retErr, i)
		}
	}

	return retResult, retErr
}

func (pm *ThriftPoolManager) GetPoolElement() *ThriftPool {
	if len(pm.poolList) < 1 {
		return nil
	}
	var pool *ThriftPool = nil
	pm.lock.Lock()
	loop := 0
	for {
		pool = pm.poolList[pm.poolCursor]
		pm.poolCursor++
		if pm.poolCursor >= len(pm.poolList) {
			pm.poolCursor = 0
		}
		if !pool.removed {
			break
		}
		loop++
		if loop >= len(pm.poolList) {
			log.Errorf("all pool is removed, will choose random")
			pool.Recover()
			pool.failNum = pm.maxFailCount - 1
			pm.poolCursor++
			if pm.poolCursor >= len(pm.poolList) {
				pm.poolCursor = 0
			}
			break
		}
	}

	pm.lock.Unlock()
	return pool
}

func (pm *ThriftPoolManager) RemovePoolElement(pool *ThriftPool) {
	if len(pm.poolList) < 1 {
		return
	}
	pm.lock.Lock()
	pool.removed = true
	pool.Release()
	pm.lock.Unlock()
}

func (p *ThriftPool) SendRpc(methodName string, args []interface{}) (interface{}, error) {
	idleClient, err := p.Get()
	if err != nil {
		log.Errorf("can not get idle client:%+v",  err)
		return nil, err
	}
	defer p.Put(idleClient)
	method := reflect.ValueOf(idleClient.Client).MethodByName(methodName)
	paramLen := method.Type().NumIn()
	for i := 0; i < 2; i++ {
		if i > 0 {
			method = reflect.ValueOf(idleClient.Client).MethodByName(methodName)
			log.Info("send rpc use new client")
		}
		if paramLen != len(args) {
			return nil, ErrInvalidParam
		}
		in := make([]reflect.Value, paramLen)
		for j := 0; j < paramLen; j++ {
			in[j] = reflect.ValueOf(args[j])
		}
		resultValues := method.Call(in)
		if len(resultValues) > 1 {
			callErr := resultValues[1].Interface()
			if callErr == nil {
				return resultValues[0].Interface(), nil
			} else {
				if i == 0 {
					log.Errorf("method %s call error:%+v, will refresh and retry!", methodName, callErr)
					newClient, refreshErr := p.RefreshClient(idleClient)
					if refreshErr != nil {
						log.Errorf("cilent refresh error")
						idleClient.destroy = true
						return nil, callErr.(error)
					} else {
						idleClient = newClient
						log.Infof("client refresh success")
					}
					continue
				} else {
					idleClient.destroy = true
					return nil, callErr.(error)
				}
			}
		}
	}

	return nil, ErrInvalidResult
}

func (pm *ThriftPoolManager) CheckRemovedPool() {
	if len(pm.poolList) < 1 {
		return
	}
	for {
		log.Info("removed pool check start")
		for _, pool := range pm.poolList {
			if !pool.removed {
				continue
			}
			log.Infof("got removed pool to init")
			err := pool.InitPool(true)
			if err == nil {
				log.Infof("pool init success")
				pool.removed = false
			}
		}
		log.Info("removed pool check end")
		time.Sleep(REMOVEDPOOLCHECKINTERVAL * time.Second)
	}
}

func (p *ThriftPool) incrFaileNum() {
	atomic.AddInt64(&p.failNum, 1)
}

func (p *ThriftPool) resetFaileNum() {
	p.failNum = 0
}

func (p *ThriftPool) makeNewClient() (*IdleClient, error) {
	idleClient, err := p.Dial(p.ip, p.port, p.connTimeout)
	if err == nil {
		//xwj add begin 2019/1/17
		client, ok := idleClient.Socket.Conn().(*net.TCPConn)
		if !ok {
			log.Errorf("It's not ok for type *net.TCPConn,idleClient.Socket.Conn()")
			return nil, errors.New("It's not ok for type *net.TCPConn")
		}
		//xwj add end 2019/1/17
		client.SetNoDelay(true)
		client.SetLinger(0)
	}
	return idleClient, err
}

func (p *ThriftPool) Get() (*IdleClient, error) {
	p.lock.Lock()
	if p.closed {
		p.lock.Unlock()
		return nil, ErrPoolClosed
	}
	if p.idle.Len() == 0 && p.count >= p.maxConn {
		p.lock.Unlock()
		return nil, ErrOverMax
	}
	log.Infof(" pool, before get, pool's idle len is:%d",  p.idle.Len())

	if p.idle.Len() == 0 {
		p.count += 1
		p.lock.Unlock()
		client, err := p.makeNewClient()
		if err != nil {
			p.lock.Lock()
			if p.count > 0 {
				p.count -= 1
			}
			p.lock.Unlock()
			return nil, err
		}
		if !client.Check() {
			p.lock.Lock()
			if p.count > 0 {
				p.count -= 1
			}
			p.lock.Unlock()
			return nil, ErrSocketDisconnect
		}
		log.Infof("pool, after get, pool's idle len is:%d",  p.idle.Len())
		return client, nil
	} else {
		ele := p.idle.Front()
		idlec, ok := ele.Value.(*IdleClient)
		if !ok {
			log.Error("It's not ok for type IdleClient")
			p.idle.Remove(ele)
			p.lock.Unlock()
			return nil, errors.New("It's not ok for type IdleClient")
		}
		p.idle.Remove(ele)
		p.lock.Unlock()
		if !idlec.Check() {
			p.lock.Lock()
			if p.count > 0 {
				p.count -= 1
			}
			p.lock.Unlock()
			return nil, ErrSocketDisconnect
		}
		log.Infof("pool, after get, pool's idle len is:%d", p.idle.Len())
		return idlec, nil
	}
}

func (p *ThriftPool) Put(client *IdleClient) error {
	if client == nil {
		return ErrInvalidConn
	}
	log.Infof("pool, before put, pool's idle len is:%d", p.idle.Len())
	p.lock.Lock()
	if p.closed {
		p.lock.Unlock()
		err := p.Close(client)
		client = nil
		return err
	}

	if p.count > p.maxConn {
		if p.count > 0 {
			p.count -= 1
		}
		p.lock.Unlock()
		err := p.Close(client)
		client = nil
		return err
	}

	if !client.Check() {
		if p.count > 0 {
			p.count -= 1
		}
		p.lock.Unlock()
		err := p.Close(client)
		client = nil
		return err
	}

	p.idle.PushBack(client)
	p.lock.Unlock()
	log.Infof(" pool, after put, pool's idle len is:%d",  p.idle.Len())

	return nil
}

func (p *ThriftPool) CheckTimeout() {
	log.Infof("before idle timeout check, pool's idle len:%d", p.idle.Len())
	p.lock.Lock()
	for p.idle.Len() > p.initConn {
		ele := p.idle.Back()
		if ele == nil {
			break
		}
		v, ok := ele.Value.(*IdleClient)
		//xwj add begin 2019/1/17
		if !ok {
			log.Errorf("It's not ok for type *IdleClient,ele.Value")
			p.idle.Remove(ele)
			break
		}
		//xwj add end 2019/1/17
		if v.CreateTime.Add(p.idleTimeout).After(nowFunc()) {
			break
		}

		p.idle.Remove(ele)
		p.lock.Unlock()
		p.Close(v) //close client connection
		p.lock.Lock()
		if p.count > 0 {
			p.count -= 1
		}
	}
	p.lock.Unlock()
	log.Infof("after idle timeout check,pool's idle len:%d", p.idle.Len())
	return
}

func (p *ThriftPool) RefreshClient(client *IdleClient) (*IdleClient, error) {
	newClient, err := p.makeNewClient()
	if err == nil {
		p.Close(client)
		return newClient, nil
	}

	return nil, err
}

func (p *ThriftPool) InitPool(reset bool) error {
	if reset {
		p.Recover()
	}

	for i := 0; i < p.initConn; i++ {
		client, err := p.makeNewClient()
		if err != nil {
			log.Error("pool init error")
			return errors.New("pool init error!!")
		}
		p.idle.PushBack(client)
		p.count += 1
	}
	return nil
}

func (c *IdleClient) LocalAddr() net.Addr {
	return c.Socket.Conn().LocalAddr()
}

func (c *IdleClient) RemoteAddr() net.Addr {
	return c.Socket.Conn().RemoteAddr()
}

func (c *IdleClient) Check() bool {
	if c.Socket == nil || c.Client == nil || c.destroy {
		return false
	}
	return c.Socket.IsOpen()
}
func (p *ThriftPool) GetPoolKey() string {
	return p.ip + "|" + p.port
}
func (p *ThriftPool) GetIdleCount() int {
	return p.idle.Len()
}

func (p *ThriftPool) GetConnCount() int {
	return p.count
}

func (pm *ThriftPoolManager) StartIdleConnClearTask() {
	if len(pm.poolList) < 1 {
		return
	}
	for {
		for _, pool := range pm.poolList {
			pool.CheckTimeout()
		}

		time.Sleep(IDLECHECKINTERVAL * time.Second)
	}
}

func (p *ThriftPool) Release() {
	p.lock.Lock()
	idle := p.idle
	p.idle.Init()
	p.closed = true
	p.count = 0
	p.resetFaileNum()
	p.lock.Unlock()
	for iter := idle.Front(); iter != nil; iter = iter.Next() {
		//xwj add begin 2019/1/17
		idleClient, ok := iter.Value.(*IdleClient)
		if !ok {
			log.Error("It's not ok for type *IdleClient,iter.Value")
			continue
		}
		//xwj add end 2019/1/17
		p.Close(idleClient)
	}
}

func (p *ThriftPool) Recover() {
	p.lock.Lock()
	p.closed = false
	p.removed = false
	p.lock.Unlock()
}
